package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.SparkSession

object _07_ES_TO_DWS {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("CheckDataDetail")
      .config("spark.debug.maxToStringFields", "2000")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.caseSensitive", "true")
      .master("local[*]")
      .getOrCreate()

    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")

    /**
     * 每月需要改日期 run
     * 然后将本地的 csv 文件 导入到数仓分区表 "test_zyf"."o2o_kuaishou_2022" 表中
     */
    val year = 2022
    val month = 11

//    spark.read.orc(s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/es/${year}/${month}/*").printSchema()

    val value: RDD[String] = spark.read.orc(s"s3a://o2o-tempdata/zyf/livestreaming/kuaishou/es/${year}/${month}/*")
      .drop("Base_Info")
      .toJSON.rdd.map(
      line => {
//        val lines = line
//          .replaceAll("\\\\\\\\r\\\\\\\\n", "")
//          .replaceAll("\\\\\\\\r", "")
//          .replaceAll("\\\\\\\\n", "")
//          .replaceAll("\\\\r\\\\n", "")
//          .replaceAll("\\\\r", "")
//          .replaceAll("\\\\n", "")
//          .replaceAll("\\r\\n", "")
//          .replaceAll("\\n", "")
//          .replaceAll("\\r", "")
//          .replaceAll("\r\n", "")
//          .replaceAll("\r", "")
//          .replaceAll("\n", "")
        val nObject: JSONObject = JSON.parseObject(line)
        val shopName = nObject.getOrDefault("shopName", "-1").toString
          .replaceAll("\"", "").replaceAll("\\\"", "")
        val title = nObject.getOrDefault("title", "-1").toString
          .replaceAll("\"", "").replaceAll("\\\"", "")
        nObject.put("shopName", shopName)
        nObject.put("title", title)
        nObject.toString
      })

    spark.read.json(value).registerTempTable("t1")
    spark.sql(
      """
        |select
        |allItemCount,
        |--Base_Info,
        |shopType,
        |brand_company_name,
        |brandValueId,
        |brandName,
        |categoryId,
        |categoryName,
        |commentCount,
        |company_name,
        |detailcategoryid,
        |detailcategoryname,
        |fans,
        |firstCategoryId,
        |fourthCategoryId,
        |good_id,
        |goodRatePercentage,
        |goodUrl,
        |images,
        |is_flagship,
        |is_market,
        |is_newProduct,
        |is_oversea,
        |is_selfSupport,
        |original_cost,
        |originCountryName,
        |originItem,
        |originProduction,
        |parentbrandid,
        |platformId,
        |price,
        |rootCategoryId,
        |rootCategoryName,
        |salesAmount,
        |secondCategoryId,
        |sellCount,
        |shopId,
        |shopUrl,
        |subCategoryId,
        |subCategoryName,
        |taxFreeZone,
        |thirdCategoryId,
        |title,
        |userId,
        |platformName,
        |timeStamp,
        |address,
        |administrative_region,
        |aedzId,
        |city,
        |city_grade,
        |city_origin,
        |district,
        |district_origin,
        |economic_division,
        |if_city,
        |if_district,
        |if_state_level_new_areas,
        |latitude,
        |longitude,
        |name,
        |poor_counties,
        |province,
        |regional_ID,
        |registration_institution,
        |rural_demonstration_counties,
        |rural_ecommerce,
        |the_belt_and_road_city,
        |the_belt_and_road_province,
        |the_yangtze_river_economic_zone_city,
        |the_yangtze_river_economic_zone_province,
        |town,
        |urban_agglomerations,
        |shopName,
        |'' as brandSourceId,
        |brand_state,
        |brand_isLaoZiHao,
        |brand_type,
        |brandName_cn,
        |'' as brandName_en
        |from t1
        |""".stripMargin)
      .repartition(1).write.mode("overwrite").option("header", true).csv(s"D:\\o2o4zyfBywork\\需求\\快手直播\\导回dws\\${year}${month}")
    //      .repartition(1).write.mode("overwrite").option("header", true).csv(s"s3a://o2o-dataproces-group/zyf/test/5/")

  }

}